package com.tomdd.message;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.tomdd.message.enums.MessageTypeEnum;
import com.tomdd.message.model.Message;
import com.tomdd.message.model.OutMessage;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 消息路由器
 *
 * @author zx
 * @date 2025年04月25日 13:56
 */
@Slf4j
public class MessageRouter {
    private static final int DEFAULT_THREAD_POOL_SIZE = 100;


    //@Getter
    //private final List<MessageRouterRule> rules = new ArrayList<>();

    @Getter
    private final Map<MessageTypeEnum, MessageRouterRule> rules = new ConcurrentHashMap<>();

    private ExecutorService executorService;

    public MessageRouter() {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("MessageRouter-pool-%d").build();
        this.executorService = new ThreadPoolExecutor(DEFAULT_THREAD_POOL_SIZE, DEFAULT_THREAD_POOL_SIZE,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), namedThreadFactory);
    }

    public MessageRouter(ExecutorService executorService) {
        this.executorService = executorService;
    }

    /**
     * 系统退出前，应该调用该方法
     */
    public void shutDownExecutorService() {
        this.executorService.shutdown();
    }

    /**
     * 系统退出前，应该调用该方法，增加了超时时间检测
     *
     * @param second the second
     */
    public void shutDownExecutorService(Integer second) {
        this.executorService.shutdown();
        try {
            if (!this.executorService.awaitTermination(second, TimeUnit.SECONDS)) {
                this.executorService.shutdownNow();
                if (!this.executorService.awaitTermination(second, TimeUnit.SECONDS)) {
                    log.error("线程池未关闭！");
                }
            }
        } catch (InterruptedException ie) {
            this.executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    //建造者设计模式
    // 实现这种效果:
    // new MessageRouter().rule().async(true).msgType("text").handler(new TextMessageHandler()).end();
    public MessageRouterRule rule() {
        return new MessageRouterRule(this);
    }

    public OutMessage route(final Message message) {
        return this.route(message, new HashMap<>(2));
    }

    public OutMessage route(final Message message, final Map<String, Object> context) {
        //注意处理重复消息问题 根据消息ID + redis 做防重
        //收集匹配规则
        final List<MessageRouterRule> matchRules = new ArrayList<>();
        MessageTypeEnum messageType = message.getMessageType();


        if (messageType == null) {
            return null;
        }
        if (this.rules.containsKey(messageType)) {
            matchRules.add(this.rules.get(messageType));
        }

        //for (MessageRouterRule rule : this.rules) {
        //    if (rule.test(message)) {
        //        matchRules.add(rule);
        //    }
        //}

        if (matchRules.isEmpty()) {
            return null;
        }
        //是否开启异步,如果开启异步处理消息。这里天极爱异步任务
        OutMessage res = null;
        //使用complatablefuture 异步多线程优化处理.
        for (MessageRouterRule rule : matchRules) {
            // 返回最后一个非异步的rule的执行结果
            if (rule.isAsync()) {
                CompletableFuture<OutMessage> outMessageCompletableFuture =
                        CompletableFuture.supplyAsync(() -> rule.service(message, context), executorService);
                try {
                    res = outMessageCompletableFuture.get();
                } catch (Exception e) {
                    log.error("异步任务执行失败", e);
                }

            } else {
                //同步操作
                res = rule.service(message, context);
            }
        }
        return res;
    }


}
